package com.qyer.search.engine;

import static com.qyer.commons.utils.CommonUtils.LEFT;
import static com.qyer.commons.utils.CommonUtils.RIGHT;
import static com.qyer.commons.utils.CommonUtils.splitList;

import com.qyer.log.WarningLogger;
import com.qyer.search.Constants;
import com.qyer.search.cache.RedisCacheHolder;
import com.qyer.search.cache.mutable.MutableCaches;
import com.qyer.search.enums.FieldsMode;
import com.qyer.search.model.SearchParameter;
import com.qyer.search.searchplan.SQPlan2;
import com.qyer.search.searchplan.SQPlan3;
import com.qyer.words.model.Word;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ThreadLocalRandom;

/**
 * Created by tangzheng on 16/8/11.
 */
public class RecursiveSearchASIDSetTask extends RecursiveTask<Set<String>> {

  private static final int TIMEOUT_IN_SECONDS = 60;
  private static final int TIMEOUT_IN_SECONDS_3 = 20;

  private AbstractESEngine searchEngine;

  private List<SQPlan3> plans;

  private boolean showHighlight;
  private FieldsMode mode;

  protected String getCacheKeyLong(SearchParameter sp, final List<Word> words) {
    long result = sp.getZCacheKey2();
    if (CollectionUtils.isEmpty(words)) {
      return "v7.test.null."+ mode + String.valueOf(result);
    }
    long result2 = 0;
    List<Word> newList = new ArrayList<>(words);
    Collections.sort(newList);
    result = 31 * result + newList.hashCode();
    //basic 是否区分 fix
    return "v7.test.so." + mode + String.valueOf(result);
  }

  public RecursiveSearchASIDSetTask(AbstractESEngine searchEngine, List<SQPlan3> plans,
                                  boolean showHighlight, FieldsMode mode) {
    if (CollectionUtils.isEmpty(plans)) {
      throw new RuntimeException("Empty plan.");
    }
    this.searchEngine = searchEngine;
    this.plans = plans;
    this.showHighlight = showHighlight;
    this.mode = mode;
  }

  public Set<String> atomicPlanSearch(SQPlan3 plan) {

    long t1 = System.currentTimeMillis();
    SearchParameter sp = plan.getSearchParameter();
    Set<String> result = new HashSet<>();
    if(plan.isSType()){
      int tid = sp.getType2();
      result = MutableCaches.getInstance().P_SNAPSHOT.getCacheContent().getSt_product_str().get(tid);
      //System.out.println("ST 命中");
    }else if(plan.isSTG()){
      int tid = sp.getType1();
      int channel = plan.getChannel();
      result = MutableCaches.getInstance().P_SNAPSHOT.getCacheContent().getStg_product_str()
                                                     .get(channel).get(tid);
      //System.out.println("STG命中");
    }else if(plan.isHotPlace()){
      //String city = sp.getDestination();
      String country = sp.getDestCountry();
      if(StringUtils.isNotEmpty(country)){
        int d = Constants.HOT_COUNTRY_MAP.get(country);
        result = MutableCaches.getInstance().P_SNAPSHOT.getCacheContent().getCountry_product().get(d);
        //System.out.println("国家命中");
      }
      /*if(StringUtils.isNotEmpty(city)){
        if(Constants.HOT_CITY_MAP.containsKey(city)){
          int d = Constants.HOT_CITY_MAP.get(city);
          result = MutableCaches.getInstance().P_SNAPSHOT.getCacheContent().getCity_product().get(d);
          System.out.println("城市命中");
        }
      }else {
        if(Constants.HOT_COUNTRY_MAP.containsKey(country)){
          int d = Constants.HOT_COUNTRY_MAP.get(country);
          result = MutableCaches.getInstance().P_SNAPSHOT.getCacheContent().getCountry_product().get(
            d);
          System.out.println("国家命中");
        }
      }*/
    } else if(plan.isOnlyContinent()){
      int continentId = MutableCaches.getInstance().CONTINENT_DICT.getCacheContent().getReverseDict().get(sp.getContinent());
      result = MutableCaches.getInstance().P_SNAPSHOT.getCacheContent().getContinent_product().get(continentId);
      //System.out.println("大洲命中");
    }
    else {
      String key = getCacheKeyLong(sp, plan.getWords());
      //System.out.println("key:" + key);

      long t8 = System.currentTimeMillis();
      List<String> result_redis = RedisCacheHolder.getInstance().getCache(key);
      long t9 = System.currentTimeMillis();

      try {
        if (CollectionUtils.isEmpty(result_redis)) {
          WarningLogger.logWarning(key + ":cache_missed", sp, null);
          //System.out.println(key + ":cache_missed");
          long t3 = System.currentTimeMillis();
          result = searchEngine.executePlanExtractAsIDSet(sp, plan, showHighlight, mode);
          long t4 = System.currentTimeMillis();
          if (CollectionUtils.isNotEmpty(result)) {
          }
          result_redis = new ArrayList<>();
          if (result != null) {
            result_redis.addAll(result);
            //int lize = 0;
            //for (String str : result_redis) {
              //lize = lize + str.getBytes().length;
            //}
            //System.out.println("result bytes length:" + lize);
          }
          long t5 = System.currentTimeMillis();
          RedisCacheHolder.getInstance().putCache(key, result_redis, (TIMEOUT_IN_SECONDS + (ThreadLocalRandom.current()
                                                                                                             .nextInt(
                                                                                                               0,
                                                                                                               TIMEOUT_IN_SECONDS_3)
                                                  )
                                                  ));
          long t6 = System.currentTimeMillis();
        } else {
          WarningLogger.logWarning(key + ":cache_hit", sp, null);
          //System.out.println(key + ":cache_hit");
          //System.out.println("redis result"+result_redis);
          result.addAll(result_redis);
          //AbstractESEngine.logDebugMap(sp, true, plan);
        }
      } catch (Exception e) {
        e.printStackTrace();
        //System.out.println(e.toString());
      }
    }
    /*if (result == null) {
      result = Collections.emptyList();
    }*/
    long t2 = System.currentTimeMillis();
    //System.out.println("atomic search time:"+(t2-t1));
    return result;
  }

  @Override protected Set<String> compute() {
    Set<String> result;
    if (CollectionUtils.isEmpty(plans)) {
      result = Collections.emptySet();
    } else if (plans.size() == 1) {
      result = atomicPlanSearch(plans.get(0));
      if(CollectionUtils.isEmpty(result)) {
        result = new HashSet<>();
        //System.out.println("ES null");
      }else {
        //System.out.println("ES ONLY TASK result size:"+result.size());
      }
    } else {
      List<SQPlan3> l = splitList(plans, LEFT);
      RecursiveSearchASIDSetTask left = new RecursiveSearchASIDSetTask(searchEngine, l, showHighlight, mode);
      l = splitList(plans, RIGHT);
      RecursiveSearchASIDSetTask right = new RecursiveSearchASIDSetTask(searchEngine, l, showHighlight, mode);
      left.fork();
      right.fork();
      int mergeSize = 0;
      Set<String> leftResult = left.join(), rightResult = right.join();
      mergeSize += CollectionUtils.isEmpty(leftResult) ? 0 : leftResult.size();
      mergeSize += CollectionUtils.isEmpty(rightResult) ? 0 : rightResult.size();
      result = new HashSet<>(mergeSize);
      if(CollectionUtils.isNotEmpty(leftResult)) {
        result.addAll(leftResult);
      }
      if(CollectionUtils.isNotEmpty(rightResult)) {
        result.addAll(rightResult);
      }
      //System.out.println("TASK result size:"+result.size());
    }
    return result;
  }
}
